package com.donglink.test.storm.wordcount;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.*;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.windowing.TupleWindow;

import java.util.HashMap;
import java.util.Map;

public class CountBolt extends BaseRichBolt {

    Map<String,Integer> map=new HashMap<>();
    OutputCollector outputCollector ;
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

        this.outputCollector=outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {

        String word=tuple.getStringByField("word");
        int count=tuple.getIntegerByField("count");

        if(map.containsKey(word))
        {

            int count1=map.get(word);
            map.put(word,count+count1);
        }else {
            map.put(word,count);
        }

        System.out.println(map);

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
